package com.flume.custom;

import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// 从配置文件中读取一个后缀，将event的内容读取后，拼接后缀进行输出
public class MySink extends AbstractSink implements Configurable {

	private String suffix;
	private Logger logger = LoggerFactory.getLogger(MySink.class);

	// 核心逻辑：处理sink逻辑
	// Satus.ready：成功传输了一个或多个event
	// Satus.backoff：从channel中无法获取数据
	@Override
	public Status process() throws EventDeliveryException {

		Status status = Status.READY;

		// 获取当前sink对接的channel
		Channel channel = getChannel();

		// 声明Event，用来接收channel中的event
		Event event = null;

		// 获取take事务对象
		Transaction tx = channel.getTransaction();

		try {

			// 开启事务
			tx.begin();

			// 如果channel中没有可用的event，此时取到的结果是null
			event = channel.take();

			if (event == null) {

				status = Status.BACKOFF;

			} else {

				// 取到数据后，执行拼接后缀进行输出
				logger.info("Event:" + new String(event.getBody()) + suffix);

			}

			// 提交事务
			tx.commit();

		} catch (ChannelException e) {

			// 异常时回滚事务，修改状态
			tx.rollback();
			status = Status.BACKOFF;

			e.printStackTrace();
		} finally {

			// 关闭事务对象
			tx.close();

		}

		return status;

	}

	// 从配置文件中读取配置的参数
	@Override
	public void configure(Context context) {

		suffix = context.getString("suffix", "!!!");

	}

}
